package dyyx;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Map;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

/**
 * 消费慢 触发重平衡 例子
 * @author dugang
 *
 */
public class ConsumerRebalanceTest {

	public static void main(String[] args) throws Exception {

		Properties props = new Properties();
		
		// partition.assignment.strategy

		// bootstrap.servers
		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaUtil.SERVERS);
		// group.id
		props.put(ConsumerConfig.GROUP_ID_CONFIG, "kafkademo_test_group");
		// enable.auto.commit
		props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
		// auto.commit.interval.ms
		props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
		// max.poll.records
		props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "3");
		
		// max.poll.interval.ms
		props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "20000");

		


		KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
		consumer.subscribe(Arrays.asList(KafkaUtil.TOPIC));
		
		System.out.println("consumer="+consumer);
		
		Collection<TopicPartition> partitions = new ArrayList<>();
		String topic = KafkaUtil.TOPIC;
		for(int i=0;i<6;i++){
			partitions.add(new TopicPartition(topic,i));
		}
		
		Map<TopicPartition, Long> beginningOffsets = consumer.beginningOffsets(partitions);
		Map<TopicPartition, Long> endOffsets = consumer.endOffsets(partitions);
		
		System.out.println("beginningOffsets="+beginningOffsets);
		System.out.println("endOffsets="+endOffsets);

		
		long pollTimeout = 100;
       
		while (true) {
			long start = System.currentTimeMillis();
			ConsumerRecords<String, String> records = consumer.poll(pollTimeout);
			long end = System.currentTimeMillis();
            long time = end - start;
            
			System.out.println("records.size="+records.count()+",time="+time);
			for (ConsumerRecord<String, String> record : records) {
				System.out.println(record);
				
			}
			// 模拟消费耗时 ， 这个时间超过  max.poll.interval.ms ，就会触发重平衡
			Thread.sleep(30000L);
		}

	}

}
